package com.myzl.coal.mq;

import cn.hutool.core.collection.CollUtil;
import com.alibaba.fastjson2.JSONArray;
import com.myzl.coal.domain.*;
import com.myzl.coal.domain.dto.RealTimeDataDTO;
import com.myzl.coal.dto.DeviceStatusDTO;
import com.myzl.coal.mongo.component.MongoComponent;
import com.myzl.coal.mongo.domain.HistoryData;
import com.myzl.coal.service.*;
import com.myzl.coal.service.impl.DemoService;
import com.myzl.coal.service.protocol.CmdService;
import com.myzl.coal.utils.Modbus4jUtils;
import com.myzl.coal.webSocket.WebSocket;
import com.myzl.common.constant.ThresholdDescriptionConstants;
import com.myzl.common.enums.*;
import com.serotonin.modbus4j.base.ModbusUtils;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.stream.Collectors;

import static com.myzl.coal.mq.MqConstant.*;

/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
@Component
@Slf4j(topic = "admin-collect")
@Data
public class DataConsumer {

    @Value("${spring.application.name}")
    private String PROJECT_NAME;
    private final IDeviceService deviceService;
    private final IDeviceStatusService deviceStatusService;
    private final CmdService cmdService;
    private final IMessageSourceService messageSourceService;
    private final IMessageSourceRelationService messageSourceRelationService;
    private final IMessageNoticeService messageNoticeService;
    private final WebSocket webSocket;
    private final IAlarmEventService alarmEventService;
    private final IRealTimeDataService realTimeDataService;
    private final MongoComponent mongoComponent;
    private final DemoService demoService;
    private final IStatisticsService statisticsService;
    private final IDoubleOrderFlowService doubleOrderFlowService;
    private final IDataStatisticsService dataStatisticsService;
    @Value("${projectId}")
    private String projectId;


    @RabbitListener(queues = "${spring.application.name}" + SPIT + REAL_TIME_DATA)
    public void realtimeDataStorage(List<RealTimeDataDTO> dataList) {
        log.info(PROJECT_NAME + SPIT + REAL_TIME_DATA + SPIT + CONSUMER_MESSAGE + dataList);
        try {
            List<RealTimeData> realTimeDataList = dataList.stream().map(e -> {
                RealTimeData realTimeData = new RealTimeData();
                BeanUtils.copyProperties(e, realTimeData);
                return realTimeData;
            }).collect(Collectors.toList());

            if (!WebSocket.webSocketMap.isEmpty()) {
                JSONArray collect = webSocket.getResult(realTimeDataList);
                webSocket.sendMessage(collect.toJSONString());
            }
            realTimeDataService.updateRealTimeDataByType(realTimeDataList);
            List<HistoryData> historyDataList = dataList.stream().map(e -> {
                HistoryData historyData = new HistoryData();
                BeanUtils.copyProperties(e, historyData);
                historyData.initBasic();
                return historyData;
            }).collect(Collectors.toList());
            mongoComponent.saveBatch(historyDataList, HistoryData.class);

            RealTimeDataDTO realTimeDataDTO = dataList.stream().findFirst().get();
            Device device = deviceService.getById(realTimeDataDTO.getDeviceId());
            Integer slaveId = device.getDeviceConfigJSONObject().getInteger("slaveId");
            DeviceTypeEnum deviceTypeEnum = DeviceTypeEnum.getValue(realTimeDataDTO.getDeviceType());

            Map<Integer, List<RealTimeDataDTO>> collect = dataList.stream().collect(Collectors.groupingBy(RealTimeDataDTO::getDictValue));

            if (Objects.equals(deviceTypeEnum,DeviceTypeEnum.DOUBLE_ORDER)&&collect.containsKey(DeviceCommandEnum.CUMULATIVE_VOLUME.getCode())){
                //LED内容刷新&更新modbus寄存器
                List<Object> values;
                log.info("start--"+System.currentTimeMillis());
                if ("yulin".equals(projectId)){
                    values = dataStatisticsService.scaleDashboard1(realTimeDataDTO.getDeviceId());
                }else {
                    values = dataStatisticsService.scaleDashboard(realTimeDataDTO.getDeviceId());
                }
                log.info("stop--"+System.currentTimeMillis());
                if(slaveId!=null){
                    Modbus4jUtils.writeHoldingRegisterByMore(slaveId, ModbusUtils.TCP_PORT,Modbus4jUtils.REGISTER_40001,values);
                }
                //分段密度，流量计算
                List<HistoryData> saveHistoryDataList = historyDataList.stream().filter(e -> e.getDictValue().equals(DeviceCommandEnum.CUMULATIVE_VOLUME.getCode())).collect(Collectors.toList());
                if (CollUtil.isNotEmpty(saveHistoryDataList)){
                    doubleOrderFlowService.handleFlow(saveHistoryDataList.get(0));
                }
            }
        } catch (Exception e) {
            log.error(REAL_TIME_DATA+"消费失败:", e);
        }
    }


    @RabbitListener(queues = "${spring.application.name}" + SPIT + RESTORE)
    public void restore(List<AlarmEvent> dataList) {
        log.info(PROJECT_NAME + SPIT + RESTORE + SPIT + CONSUMER_MESSAGE + dataList);
        try {
            dataList.forEach(e -> {
                AlarmEvent alarmEventParams = new AlarmEvent();
                alarmEventParams.setDeviceId(e.getDeviceId());
                alarmEventParams.setDictValue(e.getDictValue());
                alarmEventParams.setHandleStatus(HandleStatusEnum.UNHANDLED.getCode());
                List<AlarmEvent> alarmEvents = alarmEventService.selectAlarmEventList(alarmEventParams);
                if (CollectionUtils.isEmpty(alarmEvents)) {
                    return;
                }
                List<AlarmEvent> alarmEventList = alarmEvents.stream().map(m -> {
                    m.setHandleStatus(HandleStatusEnum.PROCESSED.getCode());
                    return m;
                }).collect(Collectors.toList());
                alarmEventService.saveOrUpdateBatch(alarmEventList);
            });
        } catch (Exception e) {
            log.error(RESTORE+"消费失败:", e);
        }

    }

    @RabbitListener(queues = "${spring.application.name}" + SPIT + POLICE)
    public void police(List<AlarmEvent> dataList) {
        log.info(PROJECT_NAME + SPIT + POLICE + SPIT + CONSUMER_MESSAGE + dataList);
        try {
            List<AlarmEvent> alarmEventList =
                    dataList.stream().filter(e -> {
                        AlarmEvent alarmEventParams = new AlarmEvent();
                        alarmEventParams.setDeviceId(e.getDeviceId());
                        alarmEventParams.setDictValue(e.getDictValue());
                        alarmEventParams.setHandleStatus(HandleStatusEnum.UNHANDLED.getCode());
                        List<AlarmEvent> alarmEvents = alarmEventService.selectAlarmEventList(alarmEventParams);
                        return CollectionUtils.isEmpty(alarmEvents);
                    }).collect(Collectors.toList());
            if (!CollectionUtils.isEmpty(alarmEventList)) {
                alarmEventService.saveOrUpdateBatch(alarmEventList);

                List<MessageSource> messageSources = messageSourceService.selectMessageSourceList(new MessageSource());
                Map<String, List<MessageSource>> sourceMap = messageSources.stream().collect(Collectors.groupingBy(MessageSource::getEventType));

                List<MessageNotice> messageNoticeList = new ArrayList<>();
                alarmEventList.forEach(e -> {
                    List<MessageSource> messageSourcesByType = sourceMap.get(e.getEventType());
                    messageSourcesByType.forEach(m -> {
                        List<MessageSourceRelation> messageSourceRelations = messageSourceRelationService.selectMessageSourceRelationList(MessageSourceRelation.builder().messageSourceId(m.getId()).build());
                        Device device = deviceService.cacheDeviceById(e.getDeviceId());
                        List<MessageNotice> messageNotices = messageSourceRelations.stream().map(map -> {
                            MessageNotice messageNotice = MessageNotice.builder()
                                    .userId(map.getUserId())
                                    .eventId(e.getId())
                                    .message(device.getDeviceName() + ":" + e.getContent())
                                    .eventType(e.getEventType())
                                    .name(m.getName())
                                    .messageSourceId(m.getId())
                                    .build();
                            return messageNotice;

                        }).collect(Collectors.toList());
                        messageNoticeList.addAll(messageNotices);
                    });
                });

                if (!CollectionUtils.isEmpty(messageNoticeList)) {
                    messageNoticeService.saveBatch(messageNoticeList);
                }
            }
        } catch (Exception e) {
            log.error(POLICE+"消费失败:", e);
        }

    }


    @RabbitListener(queues = "${spring.application.name}" + SPIT + DEVICE_STATUS)
    public void deviceStatus(DeviceStatusDTO dto) {
        log.info(PROJECT_NAME + SPIT + DEVICE_STATUS + SPIT + CONSUMER_MESSAGE + dto);

        try {

            Device device = deviceService.selectDeviceById(dto.getId());
            device.setDeviceStatus(dto.getDeviceStatus());
            deviceService.updateById(device);

            DeviceStatus deviceStatus = new DeviceStatus();

            deviceStatus.setCreateTime(new Date());
            deviceStatus.setUpdateTime(new Date());
            deviceStatus.setDeviceId(device.getId());
            deviceStatus.setDeviceType(device.getDeviceType());
            deviceStatus.setDeviceStatus(dto.getDeviceStatus());

            deviceStatusService.insertDeviceStatus(deviceStatus);

            AlarmEvent params = new AlarmEvent();
            params.setDeviceId(device.getId());
            params.setDictValue(PublicDictValueEnum.OFFLINE.getCode());
            List<AlarmEvent> alarmEvents = alarmEventService.selectAlarmEventList(params);
            if (Objects.equals(DeviceStatusEnum.OFFLINE.getCode(), dto.getDeviceStatus())) {
                realTimeDataService.updateRealTimeDeviceStatus(device.getId(), StatusStringType.YES);
                if (CollectionUtils.isEmpty(alarmEvents)) {
                    if (Objects.equals(DeviceTypeEnum.LIDAR.getCode(), device.getDeviceType())) {
                        params.setEventType(AlarmEventType.SIEVE.getCode());
                    } else {
                        params.setEventType(AlarmEventType.TAPE_SCALE.getCode());
                    }
                    params.setEventLevel(AlarmLevel.HIGH.getCode());
                    params.setTitle(ThresholdDescriptionConstants.OFFLINE_TITLE);
                    params.setContent(ThresholdDescriptionConstants.OFFLINE_CONTENT);
                    alarmEventService.insertAlarmEvent(params);
                }
            }

            if (Objects.equals(DeviceStatusEnum.ONLINE.getCode(), dto.getDeviceStatus())) {
                realTimeDataService.updateRealTimeDeviceStatus(device.getId(), StatusStringType.NO);
                if (!CollectionUtils.isEmpty(alarmEvents)) {
                    AlarmEvent alarmEvent = alarmEvents.get(0);
                    alarmEvent.setHandleStatus(HandleStatusEnum.PROCESSED.getCode());
                    alarmEventService.updateAlarmEvent(alarmEvent);
                }
            }

        } catch (Exception e) {
            log.error(DEVICE_STATUS+"消费失败:", e);
        }
    }

}
